package io.kiki.sba.registry.server.clientInterface.wrapper;

import io.kiki.sba.registry.common.model.metaserver.ClientManagerAddress.AddressVersion;
import io.kiki.sba.registry.common.model.store.BaseInfo;
import io.kiki.sba.registry.common.model.store.StoreData.DataType;
import io.kiki.sba.registry.common.model.store.Subscriber;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.common.model.wrapper.WrapperInterceptor;
import io.kiki.sba.registry.common.model.wrapper.WrapperInvocation;
import io.kiki.sba.registry.remoting.netty.Channel;
import io.kiki.sba.registry.server.clientInterface.providedata.FetchClientOffAddressService;
import io.kiki.sba.registry.server.clientInterface.push.FirePushService;
import io.kiki.sba.registry.server.clientInterface.registry.ClientInterfaceRegistry;
import io.kiki.sba.registry.server.shared.remoting.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;

import static io.kiki.sba.registry.common.model.constants.ValueConstants.CLIENT_OFF;


public class ClientOffWrapperInterceptor implements WrapperInterceptor<RegisterInvokeData, Boolean> {

    private static final Logger logger = LoggerFactory.getLogger(ClientOffWrapperInterceptor.class);

    private static final long REGISTER_PUSH_EMPTY_VERSION = 1;
    @Autowired
    protected ClientInterfaceRegistry clientInterfaceRegistry;
    @Autowired
    private FirePushService firePushService;
    @Resource
    private FetchClientOffAddressService fetchClientOffAddressService;

    @Override
    public Boolean invokeCodeWrapper(WrapperInvocation<RegisterInvokeData, Boolean> wrapperInvocation) throws Exception {
        RegisterInvokeData registerInvokeData = wrapperInvocation.getParameterSupplier().get();
        BaseInfo storeData = (BaseInfo) registerInvokeData.getStoreData();

        Url url = storeData.getClientUrl();

        AddressVersion addressVersion = fetchClientOffAddressService.getAddress(url.getIp());
        if (addressVersion != null) {
            markChannel(registerInvokeData.getChannel());
            logger.info("dataInfoId:{} ,url:{} match clientOff ips.", storeData.getDataInfoId(), url.getIp());
            if (DataType.publisher == storeData.getDataType()) {
                // match client off pub, do unpub to data, make sure the publisher remove
                try {
                    clientInterfaceRegistry.unRegister(storeData);
                } catch (Throwable e) {
                    logger.error("failed to unRegister publisher {}, source={}", storeData.getDataInfoId(), url, e);
                }
                return true;
            }

            if (DataType.subscriber == storeData.getDataType()) {
                // in some case, need to push empty to new subscriber, and stop sub
                // else, filter not stop sub
                if (clientInterfaceRegistry.isPushEmpty((Subscriber) storeData) && addressVersion.isSub()) {
                    firePushService.fireOnPushEmpty((Subscriber) storeData, clientInterfaceRegistry.getDataCenterWhenPushEmpty(), REGISTER_PUSH_EMPTY_VERSION);
                    logger.info("[clientOffSub],{},{}", storeData.getDataInfoId(), RemotingHelper.getAddressString(storeData.getClientUrl()));
                }
            }
        }
        return wrapperInvocation.proceed();
    }

    private void markChannel(io.kiki.sba.registry.api.Channel channel) {
        if (!(channel instanceof Channel)) {
            return;
        }
        Channel channel1 = (Channel) channel;
        Object value = channel1.getConnAttribute(CLIENT_OFF);
        if (!Boolean.TRUE.equals(value)) {
            channel1.setChannelAttribute(CLIENT_OFF, Boolean.TRUE);
        }
    }

    @Override
    public int getOrder() {
        return 300;
    }
}
